package com.etlmaster.datax.plugins.writer.localwriter;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import com.dp.nebula.wormhole.common.AbstractPlugin;
import com.dp.nebula.wormhole.common.interfaces.ILine;
import com.dp.nebula.wormhole.common.interfaces.ILineReceiver;
import com.dp.nebula.wormhole.common.interfaces.IWriter;
import com.etlmaster.datax.common.utils.JobDBUtil;
import com.etlmaster.datax.common.utils.ParseFileUtil;
import com.etlmaster.executor.utils.LogWriter;

public class LocalWriter extends AbstractPlugin implements IWriter {

	private String error_suffix = ".error";
	private String done_suffix = ".done";
	private File file;
	private File errorFile;
	private OutputStream out = null;
	private int successNum = 0;
	private int failedNum = 0;
	private int failLuan=0;
	private int failLong=0;
	private int failColErr=0;
	private BufferedWriter bw = null;

	private Map<String,Integer> map;
	private String nullChars = "";

	private boolean errorFlag;
	
	private int fieldNum;

	private char fieldSplit;
	private char lineSplit;
	private int fieldLength;
	@Override
	public void init() {
		
		fieldNum=Integer.parseInt(getParam().getValue(ParamKey.fieldNum));
		
		fieldSplit=getParam().getCharValue(ParamKey.fieldSplit);
		lineSplit=getParam().getCharValue(ParamKey.lineSplit);
		fieldLength=Integer.parseInt(getParam().getValue(
				ParamKey.fieldLength));
		map=new HashMap<String,Integer>();
		LogWriter.addLog("INFO","开始初始化");
		file = new File(getParam().getValue(ParamKey.fileName) + done_suffix);
		errorFile = new File(getParam().getValue(ParamKey.fileName) + error_suffix);
		
		try {
			out = new FileOutputStream(errorFile, false);
			bw = new BufferedWriter(new OutputStreamWriter(out, getParam().getValue(ParamKey.encoding)));
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	public void write(ILineReceiver lineReceiver) {
		ILine line;

		while ((line = lineReceiver.receive()) != null) {

			StringBuffer str = new StringBuffer();
			errorFlag = false;
			int len = line.getFieldNum();
			int lineLength = 0;
			
			for (int i = 0; i < len; i++) {
				String ls = replaceChars(line.getField(i)).toString();
				lineLength += ls.length();
				str.append(ls);
				str.append(fieldSplit);
			}
			str.deleteCharAt(str.length() - 1);
			str.append(lineSplit);
			
			if(!(len ==fieldNum)){
				//map.put("列数不符行数", map.get("列数不符行数")+1);
				writeLine(str.toString(), errorFile);
				failColErr=failColErr+1;
				failedNum++;
				continue;
			}
			


			
			errorFlag = ParseFileUtil.isMessyCode(str.toString());
			if(errorFlag){
				//map.put("乱码行数", map.get("乱码行数")+1);
				writeLine(str.toString(), errorFile);
				failedNum++;
				failLuan=failLuan+1;
				continue;
			}

			if(lineLength > fieldLength){
				//map.put("超长行数", map.get("超长行数")+1);
				writeLine(str.toString(), errorFile);
				failedNum++;
				failLong=failLong+1;
				continue;
			}
			
			successNum++;
			
				
			}

	}

	private String replaceChars(String str) {
		if (null == str) {
			return this.nullChars;
		}
		return str;
	}

	public void writeLine(String str, File file) {
		try {

			bw.write(str);
			bw.flush();
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public void commit() {
		LogWriter.addLog("INFO","成功行数：" + successNum + ",失败行数：" + failedNum);

		
		map.put("1.成功行数", successNum);
		map.put("2.乱码行数", failLuan);
		map.put("3.超长行数", failLong);
		map.put("4.列数不符行数", failColErr);

		SimpleDateFormat sd=new SimpleDateFormat("yyyyMMdd");
		java.util.Iterator<String> it=map.keySet().iterator();
		while (it.hasNext()) {
			String key=it.next();
			String insertLog = "insert into inter_file_check_log values('"+getParam().getValue(ParamKey.fileName)+"','"
					+ key
					+ "',"
					+ map.get(key)
					+ ",current_timestamp"
					+",'"
					+sd.format(new Date())
					+ "')";
			JobDBUtil.insert(insertLog);
		}
		closeAll();
	}

	public void closeAll() {
		try {
			if (bw != null)
				bw.close();
			if (out != null)
				out.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
